package Tailfile

import (
	"context"
	"fmt"
	"logagent/Kafka"
	"logagent/common"
	"strings"
	"time"

	"github.com/Shopify/sarama"
	"github.com/hpcloud/tail"
)

type tailTask struct {
	path     string
	topic    string
	instance *tail.Tail
	ctx      context.Context
	cancel   context.CancelFunc
}

var (
	confChan chan []common.CollectEntry
)

func newTailTask(path, topic string) *tailTask {
	ctx, cancel := context.WithCancel(context.Background())
	tt := tailTask{
		path:   path,
		topic:  topic,
		ctx:    ctx,
		cancel: cancel,
	}
	return &tt

}
func (t *tailTask) Init() (err error) {
	config := tail.Config{
		ReOpen:    true,
		Follow:    true,
		Location:  &tail.SeekInfo{Offset: 0, Whence: 2},
		MustExist: false,
		Poll:      true,
	}
	t.instance, err = tail.TailFile(t.path, config)
	return
}

func (t *tailTask) run() {
	//read log and send to Kafka
	fmt.Printf("path:%s and topic:%s is working for collect log\n", t.path, t.topic)
	for {
		select {
		case <-t.ctx.Done():
			fmt.Printf("path :%v is receive message to stop \n", t.path)
			t.instance.Stop()
			return
		default:
			line, ok := <-t.instance.Lines
			if !ok {
				fmt.Printf("tail file close reopen, path:%s \n", t.path)
				time.Sleep(time.Second)
				continue
			}
			// take code to sync
			if len(strings.Trim(line.Text, "\r")) == 0 {
				fmt.Println("it have null ")
				continue
			}
			msg := &sarama.ProducerMessage{}
			msg.Topic = t.topic
			msg.Value = sarama.StringEncoder(line.Text)
			Kafka.MsgChan <- msg
		}

	}
}
